/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.scanner.ScannerFactory;
import com.nhncorp.neptune.client.scanner.TableScanner;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.NeptuneLock;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.RPCInterruptedException;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.DataServiceProtocol;
import com.nhncorp.neptune.tabletserver.Tablet;
import com.nhncorp.pleiades.client.LockService;

/**
 * Cache는 다음과 같은 레코드 구조를 가진다.
 * cahce가 아무것도 없는 상태에서 rk10에 대해 lookup 요청을 하면
 * cache에 없기 때문에 ROOT, META 정보를 이용하여 Tablet 정보를 가져온다.
 * 가져온 Tablet 정보가 T1에 max rk가 rk20인 경우 cache에는 다음 두개의 데이터가 입력된다.
 * rk10 - T1
 * rk20 - T1
 * 
 * 이렇게 계속 진행된 경우
 * 
 * rk01 - T0
 * rk05 - T0
 * rk10 - T1
 * rk20 - T1
 * 이 상태에서 rk15에 대한 요청이 오면 headSet, tailtSet한 다음 두개의 Tablet이 동일한 경우 cache hit로 처리한다.
 * rk07에 대한 요청이 온 경우 head, tail의 값이 T0, T1이기 때문에 cache miss로 처리한다.
 * 
 * ROOT, META preLoad 하는 경우 두개의 레코드를 입력하는데 start rk 값은 이전 tablet의 end rk에 1byte 추가한 
 * 값을 이용한다.
 * @author nhn
 *
 */
public class TabletLocationCache {
  private static final Log LOG = LogFactory.getLog(TabletLocationCache.class.getName());
  
  private static TabletLocationCache instance;
  
  private NConfiguration conf;
  
  private TabletInfo rootTablet;
  
  private DataServiceProtocol rootTabletServer;
  
  //TableName -> TableInfo
  private Map<String, TableSchema> tableSchemas = new HashMap<String, TableSchema>();

  protected TreeMap<Row.Key, TabletInfo> rootCache = new TreeMap<Row.Key, TabletInfo>();

  protected TreeMap<Row.Key, TabletInfo> metaCache = new TreeMap<Row.Key, TabletInfo>();

  private final NeptuneLock lock = new NeptuneLock();

  private final NeptuneLock cacheLock = new NeptuneLock();

  private LockService lockService;  
  public synchronized static TabletLocationCache getInstance(NConfiguration conf) throws IOException {
    if(instance == null) {
      instance = new TabletLocationCache(conf);
      
      try {
        instance.initRootTablet();
        instance.preLoadFromMetaTablet();
      } catch(IOException e) {
        LOG.warn("cannot load ROOT tablet or META tablet info:" + e.getMessage(), e);
      }
    }
    return instance;  
  }
  
  private TabletLocationCache(NConfiguration conf) throws IOException {
    this.conf = conf;
    this.lockService = NTableManager.getLockService(conf);
  }
  
  public Map<String, TableSchema> getTableInfos() {
    return tableSchemas;
  }
  
  protected TableSchema getTableSchema(String tableName) throws IOException {
    lock.obtainWriteLock();
    try {
      if (tableSchemas.containsKey(tableName)) {
        return tableSchemas.get(tableName);
      }
  
      TableSchema tableSchema = TableSchema.loadTableSchema(conf, lockService, tableName);
      if (tableSchema == null) {
        return null;
      }
      tableSchemas.put(tableName, tableSchema);
      return tableSchema;
    } finally {
      lock.releaseWriteLock();
    }
  }
  
  public void clearAllCache() {
    cacheLock.obtainWriteLock();
    try {
      rootTablet = null;
      rootCache.clear();
      metaCache.clear();
    } finally {
      cacheLock.releaseWriteLock();
    }
    
    lock.obtainWriteLock(); 
    try {
      tableSchemas.clear();
    } finally {
      lock.releaseWriteLock(); 
    }
  }

  public void clearTabletCache(String tableName, TabletInfo tabletInfo) {
    if(tabletInfo == null) {
      return;
    }
    cacheLock.obtainWriteLock();
    try {
      if(Constants.TABLE_NAME_ROOT.equals(tableName)) {
        rootTablet = null;
      } else if(Constants.TABLE_NAME_META.equals(tableName)) {
        rootCache.clear();
        metaCache.clear();
      } else {
        TreeMap<Row.Key, TabletInfo> tempMetaCache = new TreeMap<Row.Key, TabletInfo>();
        tempMetaCache.putAll(metaCache);
        for(Map.Entry<Row.Key, TabletInfo> entry: tempMetaCache.entrySet()) {
          Row.Key rowKey = entry.getKey();
          TabletInfo entryTabletInfo = entry.getValue();
          if(entryTabletInfo.equals(tabletInfo)) {
            metaCache.remove(rowKey);
          }
        }
      }
    } finally {
      cacheLock.releaseWriteLock();
    }
  }
  
  public void clearTabletCache(String tableName, Row.Key rowKey, TabletInfo tabletInfo) {
    cacheLock.obtainWriteLock();
    try {
      if(Constants.TABLE_NAME_ROOT.equals(tableName)) {
        rootTablet = null;
      } else if(Constants.TABLE_NAME_META.equals(tableName)) {
        if(tabletInfo == null) {
          //LOG.debug("clear all root cache");
          rootCache.clear();
        } else {
          Row.Key rootTableRowKey = Tablet.generateMetaRowKey(Constants.TABLE_NAME_META, rowKey);
          int removedCount = removeFromCache(rootCache, rootTableRowKey, tabletInfo);
          //LOG.debug("clear root cache: removedCount=" + removedCount + "," + rootTableRow.Key + "," + tabletInfo);
        }
      } else {
        if(tabletInfo == null) {
          //LOG.debug("clear all meta cache");
          metaCache.clear();
        } else {
          Row.Key metaTableRowKey = Tablet.generateMetaRowKey(tableName, rowKey);
          int removedCount = removeFromCache(metaCache, metaTableRowKey, tabletInfo);
          //LOG.debug("clear root cache: removedCount=" + removedCount + "," + metaTableRow.Key + "," + tabletInfo);
        }
      }
    } finally {
      cacheLock.releaseWriteLock();
    }
  }
  
  protected void removeTableSchemaCache(String tableName) {
    lock.obtainWriteLock();
    try {
      tableSchemas.remove(tableName);
    } finally {
      lock.releaseWriteLock();
    }
  }

  protected TableSchema reloadTableSchema(String tableName) throws IOException {
    lock.obtainWriteLock();
    try {
      tableSchemas.remove(tableName);
    } finally {
      lock.releaseWriteLock();
    }
    return getTableSchema(tableName);
  }
  
  protected TabletInfo lookupRootTablet() throws IOException {
    initRootTablet();
    return rootTablet;
  }
  
  /**
   * ROOT Tablet에서 META Tablet 정보를 가져온다.
   * @param tableName
   * @param rowKey
   * @return
   * @throws IOException
   */
  protected TabletInfo lookupMetaTablet(String tableName, Row.Key rowKey) throws IOException {
    Row.Key rootTableRowKey = Tablet.generateMetaRowKey(Constants.TABLE_NAME_META, rowKey);
    //cache에서 찾는다.
    TabletInfo tabletInfo = findFromCache(rootCache, rootTableRowKey);
    if(tabletInfo != null) {
      //Cache에 있는 경우
      return tabletInfo;
    }
    
    //없으면 직접 TabletServer로 요청
    if(tabletInfo == null) {
      //LOG.debug("Root Cache miss:" + tableName + "," + rowKey);
      initRootTablet();
      if(rootTabletServer == null) {
        return null;
      }
      try {
        //ROOT Tablet으로부터 META Tablet 정보를 가져온다.
        ColumnValue columnValue = rootTabletServer.getCloestMetaData(rootTablet.getTabletName(), rootTableRowKey);
        if(columnValue == null) {
          clearTabletCache(rootTablet.getTableName(), rootTableRowKey, rootTablet);
          return null;
        }
        tabletInfo = new TabletInfo();
        tabletInfo.readFields(columnValue.getValue());
        
        addCache(rootCache, rootTableRowKey,
            tabletInfo,
            Tablet.generateMetaRowKey(Constants.TABLE_NAME_META, tabletInfo.getEndRowKey())
            );
      } catch (Exception e) {
        //ROOT를 관리하는 TabletServer에 연결 중 장애가 발생하면 NULL을 반환
        //null을 받은 쪽에서 다시 retry할 때 다시 정보를 가져온다.
        clearTabletCache(rootTablet.getTableName(), rootTableRowKey, rootTablet);
        return null;
      }
    }
    return tabletInfo;
  }

  protected TabletInfo lookupUserTablet(String tableName, Row.Key rowKey) throws IOException {
    Row.Key metaTableRowKey = Tablet.generateMetaRowKey(tableName, rowKey);
    TabletInfo tabletInfo = findFromCache(metaCache, metaTableRowKey);
    
    if(tabletInfo == null) {
      //LOG.debug("Cache missed:" + tableName + "," + rowKey + "," + metaTableRowKey);
      //META Tablet을 먼저 찾는다.
      TabletInfo metaTablet = lookupMetaTablet(Constants.TABLE_NAME_META, metaTableRowKey);

      if(metaTablet == null) {
        //root cache clear
        clearTabletCache(Constants.TABLE_NAME_META, metaTableRowKey, null);
        LOG.warn("Cannot find meta tablet in looking up user tablet");
        return null;
      }
      
      //UserTablet을 찾는다.
      try {
        DataServiceProtocol metaTabletServer = NTableManager.connectTabletServer(metaTablet, conf);
        ColumnValue columnValue = metaTabletServer.getCloestMetaData(metaTablet.getTabletName(), metaTableRowKey);

        if(columnValue == null) {
          clearTabletCache(Constants.TABLE_NAME_META, metaTableRowKey, metaTablet);
          return null;
        }
        tabletInfo = new TabletInfo();
        tabletInfo.readFields(columnValue.getValue());

        if(tableName.equals(tabletInfo.getTableName())) {
          addCache(metaCache, metaTableRowKey, tabletInfo,
              Tablet.generateMetaRowKey(tableName, tabletInfo.getEndRowKey()));
          return tabletInfo;
        } else {
          clearTabletCache(Constants.TABLE_NAME_META, metaTableRowKey, metaTablet);
          return null;
        }
      } catch (RPCInterruptedException e) {
        return null;
      } catch (Exception e) {
        LOG.warn("Exception in reading scanner to meta", e);
        clearAllCache();
        return null;
      }
    } else {
      if(tableName.equals(tabletInfo.getTableName())) {
        return tabletInfo;
      } else {
        clearTabletCache(Constants.TABLE_NAME_META, metaTableRowKey, null);
        return null;
      }
    }
  }
  
  private void addCache(TreeMap<Row.Key, TabletInfo> cache, Row.Key lookupRowKey, 
      TabletInfo tabletInfo, Row.Key cacheEndRowKey) {
    //LOG.info("Added to cache:" + lookupRow.Key + "," + tabletInfo);
    cacheLock.obtainWriteLock();
    try {
      SortedMap<Row.Key, TabletInfo> tailMap = cache.tailMap(lookupRowKey);
      
      if(tailMap.size() == 0) {
        //LOG.info("Added to cache1:" + lookupRow.Key + "," + cacheEndRow.Key + "," + tabletInfo);
        cache.put(lookupRowKey, tabletInfo);
        cache.put(cacheEndRowKey, tabletInfo);
      } else {
        //Map.Entry<Row.Key, TabletInfo> tailFirst = tailMap.firstEntry();
        Row.Key tailFirstKey = tailMap.firstKey();
        TabletInfo tailFirstValue = tailMap.get(tailFirstKey);
        if(tailFirstValue.equals(tabletInfo) &&
            lookupRowKey.compareTo(tailFirstKey) < 0) {
          //LOG.info("Removed to cache2:" + tailMap.firstKey() + "," + tabletInfo);
          //LOG.info("Added to cache2:" + lookupRow.Key + "," + tabletInfo);
          cache.remove(tailMap.firstKey());
          cache.put(lookupRowKey, tabletInfo);
        } else {
          //LOG.info("Added to cache3:" + lookupRow.Key + "," + cacheEndRow.Key + "," + tabletInfo);
          cache.put(lookupRowKey, tabletInfo);
          cache.put(cacheEndRowKey, tabletInfo);
        }
      }
    } finally {
      cacheLock.releaseWriteLock();
    }
  }
  
  private void initRootTablet() throws IOException {
    cacheLock.obtainWriteLock();
    try {
      if(rootTablet == null) {
        rootTablet = Tablet.getRootTabletInfo(conf, lockService);
        rootTabletServer = NTableManager.connectTabletServer(rootTablet, conf);
        if(!rootTabletServer.isServicedTablet(rootTablet.getTabletName())) {
          LOG.info("Invalid ROOT tablet server:" + rootTablet);
          rootTablet = null;
          rootTabletServer = null;
          return;
        }
      } 
    } catch (IOException e) {
      rootTablet = null;
      rootTabletServer = null;
      throw e;
    } finally {
      cacheLock.releaseWriteLock();
    }    
    preLoadFromRootTablet();
  }
  
  /**
   * ROOT Tablet에 있는 META Tablet 정보를 미리 로딩 한다.
   * @throws IOException
   */
  private void preLoadFromRootTablet() throws IOException {
    TableScanner scanner = null;
    
    try {
      scanner = ScannerFactory.openScanner(conf, rootTablet, Constants.META_COLUMN_NAME_TABLETINFO);
      ScanCell scanCell = null;
      
      int count = 0;
      Row.Key previousEndRowKey = null;
      
      while( (scanCell = scanner.next()) != null ) {
        TabletInfo tabletInfo = new TabletInfo();
        tabletInfo.readFields(scanCell.getBytes());
        
        Row.Key lookupRowKey = null;
        Row.Key metaFormatRowKey = Tablet.generateMetaRowKey(Constants.TABLE_NAME_META, tabletInfo.getEndRowKey());
        if(previousEndRowKey == null) {
          lookupRowKey = Row.Key.MIN_KEY;
        } else {
          byte[] bytes = previousEndRowKey.getBytes();
          byte[] rowKeyBytes = new byte[bytes.length + 1];
          System.arraycopy(bytes, 0, rowKeyBytes, 0, bytes.length);
          rowKeyBytes[bytes.length] = 0;
          lookupRowKey = new Row.Key(rowKeyBytes);
        }
        addCache(rootCache, lookupRowKey,
            tabletInfo,
            metaFormatRowKey);
        
        previousEndRowKey = metaFormatRowKey;
        count++;
        if(count > 1000) {
          break;
        }
      }
    } finally {
      if(scanner != null) {
        scanner.close();
      }
    }
  }
  
  /**
   * 클라이언트가 처음 로딩될때 META에 있는 일부 정보(약 10000건 정도)를 로딩 시킨다.
   * @throws IOException
   */
  private void preLoadFromMetaTablet() throws IOException {
    List<TabletInfo> rootTablets = new ArrayList<TabletInfo>();
    cacheLock.obtainReadLock();
    try {
      rootTablets.addAll(rootCache.values());
    } finally {
      cacheLock.releaseReadLock();
    }
    
    int count = 0;
    for(TabletInfo eachRootTablet: rootTablets) {
      TableScanner scanner = null;
      
      try {
        scanner = ScannerFactory.openScanner(conf, 
            eachRootTablet, Constants.META_COLUMN_NAME_TABLETINFO);
        ScanCell scanCell = null;
        
        Row.Key previousEndRowKey = null;
        while( (scanCell = scanner.next()) != null ) {
          TabletInfo tabletInfo = new TabletInfo();
          tabletInfo.readFields(scanCell.getBytes());
          
          Row.Key metaFormatRowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
          Row.Key lookupRowKey = null;
          if(previousEndRowKey == null) {
            lookupRowKey = Row.Key.MIN_KEY;
          } else {
            byte[] bytes = previousEndRowKey.getBytes();
            byte[] rowKeyBytes = new byte[bytes.length + 1];
            System.arraycopy(bytes, 0, rowKeyBytes, 0, bytes.length);
            rowKeyBytes[bytes.length] = 0;
            lookupRowKey = new Row.Key(rowKeyBytes);
          }
          addCache(metaCache, lookupRowKey,
              tabletInfo,
              metaFormatRowKey);
          
          previousEndRowKey = metaFormatRowKey;          
          
          count++;
          if(count > 10000) {
            break;
          }
        }
      } finally {
        if(scanner != null) {
          scanner.close();
        }
      }
    }
  }
  
  private int removeFromCache(TreeMap<Row.Key, TabletInfo> cache, Row.Key lookupRowKey, TabletInfo removeTablet) {
    int removedCount = 0;
    SortedMap<Row.Key, TabletInfo> headMap = cache.headMap(lookupRowKey);
    if(headMap.size() > 0) {
      Row.Key headLast = headMap.lastKey();
      TabletInfo headLastValue = headMap.get(headLast);
      //System.out.println("12>>>>>" + headMap.lastEntry().getValue() + ">" + removeTablet);
      if(headLastValue.equals(removeTablet)) {
        cache.remove(headLast);
        removedCount++;
      }
    }
    SortedMap<Row.Key, TabletInfo> tailMap = cache.tailMap(lookupRowKey);
    if(tailMap.size() > 0) {
      Row.Key tailFirst = tailMap.firstKey();
      TabletInfo tailFirstValue = tailMap.get(tailFirst);      
      //System.out.println("34>>>>>" + tailMap.firstEntry().getValue() + ">" + removeTablet);
      if(tailFirstValue.equals(removeTablet)) {
        cache.remove(tailFirst);
        removedCount++;
      }
    }
    
    return removedCount;
  }
  
  protected TabletInfo findFromCache(TreeMap<Row.Key, TabletInfo> cache, Row.Key lookupRowKey) {
    cacheLock.obtainReadLock();
    try {
      if(cache.containsKey(lookupRowKey)) {
        //LOG.debug("findFromCache:match rowkey:" + lookupRow.Key);
        return cache.get(lookupRowKey);
      }
      SortedMap<Row.Key, TabletInfo> headMap = cache.headMap(lookupRowKey);
      //LOG.debug("findFromCache:headMap:" + lookupRow.Key + "," + headMap.size());
      if(headMap.size() == 0) {
        return null;
      }
      SortedMap<Row.Key, TabletInfo> tailMap = cache.tailMap(lookupRowKey);
      //LOG.debug("findFromCache:tailMap:" + lookupRow.Key + "," + tailMap.size());
      if(tailMap.size() == 0) {
        return null;
      }
      
      Row.Key headLast = headMap.lastKey();
      Row.Key tailFirst = tailMap.firstKey();
      
      TabletInfo headLastValue = headMap.get(headLast);
      TabletInfo tailFirstValue = tailMap.get(tailFirst);
      
      if(lookupRowKey.compareTo(headLast) >= 0 
          && lookupRowKey.compareTo(tailFirst) <= 0
          && headLastValue.equals(tailFirstValue)) { 
        return  tailFirstValue;
      } else {
        return null;
      }
    } finally {
      cacheLock.releaseReadLock();
    }
  }
}
